package com.myCompany.edu.realtime.util;

import com.myCompany.edu.realtime.common.Constant;

/**
 * @ClassName：SQLUtil
 * @Author: xxx
 * @Date: 2022/10/20 14:57
 * @Description:
 */
public class SQLUtil {

    public static String getKafkaSourceSQL(String topic, String groupId) {
        return "with(" +
                "'connector' = 'kafka', " +
                "'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS + "', " +
                "'properties.group.id' = '" + groupId + "', " +
                "'topic' = '" + topic + "', " +
                "'format' = 'json', " +
                "'scan.startup.mode' = 'latest-offset' " +
                ")";
    }
    public static String getKafkaSinkSQL(String topic) {
        return "with(" +
                "'connector' = 'kafka', " +
                "'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS + "', " +
                "'topic' = '" + topic + "', " +
                "'format' = 'json' " +
                ")";
    }

    public static String getUpsertKafkaSinkSQL(String topic) {
        return "with(" +
                "'connector' = 'upsert-kafka', " +
                "'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS + "', " +
                "'topic' = '" + topic + "', " +
                "'key.format' = 'json', " +
                "'value.format' = 'json' " +
                ")";
    }

}
